[drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect #21351
Conversation
poorbarcode
changed the title
| @@ -2284,12 +2285,18 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long e | |||
| if (stripChecksum) { | |||
| stripChecksum(op); | |||
| } | |||
| // To avoid IO buffer overflow, split to multi-flush. | |||
| if (messageBytesSizeInCache + op.cmd.readableBytes() < messageBytesSizeInCache) { | |||
| cnx.ctx().flush(); | |||
There was a problem hiding this comment.
flush is an async operation, so this might not work as expected
There was a problem hiding this comment.
Since write and flush will be executed in the same thread, the actions of both write and flush will keep the order to execute. So it will work as expected.
There was a problem hiding this comment.
I mean from a different aspect. I guess that in this case, the purpose of flushing is to ensure that buffers don't overflow. It feels like the solution in this PR won't fully address that since all operations will be queued up in any case unless the logic that calls write is also run in the ctx loop (thread). Other possibility would be to somehow wait for flush completion until it finishes.
There was a problem hiding this comment.
In Netty, there's the Channel.isWritable method and channelWritabilityChanged callback that help in keeping the queued writes bounded (see low/high watermark options). However, IIRC, Pulsar code base doesn't show examples of how write logic could be implemented to take advantage of this type of backpressure solution.
Optimally, the logic would be implemented in a way where more writes are added while the channel is writable and then paused. Adding more writes should resume after the channel becomes writable again.
There was a problem hiding this comment.
I don't think that flushing will help resolve the issue.
In Netty, there's the Channel.isWritable method and channelWritabilityChanged callback that help in keeping the queued writes bounded (see low/high watermark options). However, IIRC, Pulsar code base doesn't show examples of how write logic could be implemented to take advantage of this type of backpressure solution.
Optimally, the logic would be implemented in a way where more writes are added while the channel is writable and then paused. Adding more writes should resume after the channel becomes writable again.
|
I'm still not sure how it gets to 2GB of accumulated data. The client memory limit backpressure should have blocked long before the 2GB. |
Motivation
If too many messages were called
sendAsyncwhen the state of the producer is reconnecting, after connection successfully, the messages cached in the memory will be flushed to the socket once, then the buffer overflow.Modifications
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: x